package dataStream;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc connect 可以连接两个不同类型的流,分别进行操作之后转换为同一类型
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class ConnectStreamFunc {
    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        List<String> data2 = new ArrayList<>();
        data2.add("xu");
        data2.add("xuyou");
        data2.add("xu2");
        data2.add("xu3");
        data2.add("xu4");


        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream1 = env.fromCollection(data);
        DataStream<String> dataStream2 = env.fromCollection(data2);

        dataStream1.connect(dataStream2).process(new CoProcessFunction<Person, String, Object>() {
            @Override
            public void processElement1(Person value, Context ctx, Collector<Object> out) throws Exception {

            }

            @Override
            public void processElement2(String value, Context ctx, Collector<Object> out) throws Exception {

            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
            }
        });

        SingleOutputStreamOperator<Object> connectStream = dataStream1.connect(dataStream2).map(new CoMapFunction<Person, String, Object>() {
            @Override
            public Object map1(Person person) throws Exception {
                return person.getGender() + person.getAge();
            }

            @Override
            public Object map2(String s) throws Exception {
                return s + "_abc";
            }
        });

        connectStream.print();
        env.execute();
    }
}
